package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: gaojian
 * @Description: 针对企查查t_tax_owe表和中数t_tax_arrears表进行数据映射开发
 * @Date: 2021/12/29
 */

public class TTaxOwe {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_tax_owe").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_tax_arrears");
        long zs_count = zs_data.count();
        if(zs_count>0){

            // 处理步骤说明 1.UPDATE 2.INSERT 3.DELETE
            // 1.UPDATE
            // SQL逻辑：定义更新集合 update_data
            //          1.从 dc_ep_ods 取得中数表全集 统一社会信用代码 不为空 为集合A
            //          2.从 xy_ods 取得企查查全集 统一社会信用代码 不为空 为集合B
            //          3.集合A 和 集合B 通过统一社会信用代码，做INNER JOIN，注意要 UPPER 和 TRIM 函数 为集合C
            //              key_no，company_id 来自企查查
            //              JOB_ID, "0" as isadd 两个字段是要增加的
            //          4.从 dc_ep_ods 取得中数表全集 组织机构号 不为空 并且 统一社会信用代码为空的 为集合D
            //          5.从 xy_ods 取得企查查全集 组织机构号 不为空 为集合E
            //          6.集合D 和 集合E 通过组织机构号，做INNER JOIN 注意要 UPPER 和 TRIM 函数 为集合F
            //              key_no，company_id 来自企查查
            //              JOB_ID, "0" as isadd 两个字段是要增加的
            //          7.集合C 与 机会F 使用 UNION 为并集 G
            //          8.update_data.createOrReplaceTempView 更新
            //          9.同步到目标表

            Dataset<Row> update_data = spark.sql(
                    "SELECT \n" +
                            "B.ID,\n" +
                            "B.KEY_NO,\n" +
                            "B.COMPANY_ID,\n" +
                            "B.COMPANY_NAME,\n" +
                            "B.TAX_OWE_ID,\n" +
                            "A.TAX_CATEGORY AS TITLE,\n" +
                            "A.NOTICE_DATE AS PUBLISH_DATE,\n" +
                            "A.COMPETENT_TAX_AUTHORITY AS ISSUED_BY,\n" +
                            "A.BALANCE AS AMOUNT,\n" +
                            "A.BALANCE_NEW AS NEW_AMOUNT,\n" +
                            "A.JOBID,\n" +
                            "'0' AS ISADD \n" +
                            "FROM \n" +
                            "(\n" +
                            "    select del.creditcode,t.* \n" +
                            "    from a_dc_ep_ods.zs_t_tax_arrears t  \n" +
                            "    inner join a_dc_ep_ods.zs_t_tax_arrears_del del on del.entid = t.entid\n" +
                            "    where del.creditcode is not null and del.creditcode != '' \n" +
                            ") A INNER JOIN \n" +
                            "(\n" +
                            "    select c.credit_code,c.no as regno,o.*\n" +
                            "    from a_dc_ep_dwi.t_tax_owe o\n" +
                            "    inner join a_dc_ep_ods.t_eci_company c on c.company_id = o.company_id\n" +
                            "    where c.credit_code is not null and c.credit_code != '' \n" +
                            ") B \n" +
                            "ON UPPER(TRIM(B.CREDIT_CODE)) = UPPER(TRIM(A.CREDITCODE)) \n" +
                            "WHERE \n" +
                            "DATE_FORMAT(IF(A.NOTICE_DATE = '' OR A.NOTICE_DATE IS NULL,'1900-01-01',A.NOTICE_DATE),'yyyy-MM-dd')\n" +
                            "=\n" +
                            "DATE_FORMAT(IF(B.PUBLISH_DATE = '' OR B.PUBLISH_DATE IS NULL,'1900-01-01',B.PUBLISH_DATE),'yyyy-MM-dd')\n" +
                            "AND UPPER(TRIM(A.TAX_CATEGORY)) = UPPER(TRIM(B.TITLE))");
            update_data.createOrReplaceTempView("tmp_t_tax_owe");
            spark.sql("insert overwrite table a_dc_ep_incr.t_tax_owe select * from tmp_t_tax_owe");

            // 2.INSERT
            // SQL逻辑：定义 insert_data 集合
            //          1.从 dc_ep_ods 取得中数表全集 统一社会信用代码 不为空 为集合A
            //          2.从 xy_ods 取得企查查全集 统一社会信用代码 不为空 为集合B
            //          3.集合A 和 集合B 通过统一社会信用代码，做LEFT JOIN，注意要 UPPER 和 TRIM 函数，再筛选企查查侧为空的 为集合C
            //                  产生 key_no 和 company_id 使用 md5(concat(集合A.entname,if(length(集合A.creditcode)=18,集合A.creditcode,'')))
            //                  JOB_ID, "1" as isadd 两个字段是要增加的
            //          4.从 dc_ep_ods 取得中数表全集 组织机构号 不为空 并且 统一社会信用代码为空的 为集合D
            //          5.从 xy_ods 取得企查查全集 组织机构号 不为空 为集合E
            //          6.集合D 和 集合E 通过组织机构号，做 LEFT JOIN 注意要 UPPER 和 TRIM 函数，再筛选企查查侧为空的 为集合F
            //                  产生 key_no 和 company_id 使用 md5(concat(集合A.entname))
            //                  JOB_ID, "1" as isadd 两个字段是要增加的
            //          7.集合C 与 机会F 使用 UNION 为并集 G
            //          8.insert_data.createOrReplaceTempView 完成插入
            //          9.同步到目标表
            Dataset<Row> insert_data = spark.sql(
                    "SELECT\n" +
//                            "md5(concat(UPPER(TRIM(A.TAXPAYER_NAME)),A.CREDITCODE,DATE_FORMAT(IF(A.NOTICE_DATE = '' OR A.NOTICE_DATE IS NULL,'1900-01-01',A.NOTICE_DATE),'yyyy-MM-dd'))) AS ID,\n" +
                            "A.RECORD_ID AS ID,\n" +
                            "if(B.company_id is not null,B.company_id,md5(concat(A.TAXPAYER_NAME,if(length(A.creditcode)=18,A.creditcode,'')))) AS KEY_NO, \n" +
                            "if(B.company_id is not null,B.company_id,md5(concat(A.TAXPAYER_NAME,if(length(A.creditcode)=18,A.creditcode,'')))) AS COMPANY_ID, \n" +
                            "A.TAXPAYER_NAME AS COMPANY_NAME,\n" +
                            "md5(concat(UPPER(TRIM(A.TAXPAYER_NAME)),A.CREDITCODE,\n" +
                            "DATE_FORMAT(IF(A.NOTICE_DATE = '' OR A.NOTICE_DATE IS NULL,'1900-01-01',A.NOTICE_DATE),'yyyy-MM-dd'),UPPER(TRIM(A.TAX_CATEGORY)))) AS TAX_OWE_ID,\n" +
                            "A.TAX_CATEGORY AS TITLE,\n" +
                            "A.NOTICE_DATE AS PUBLISH_DATE,\n" +
                            "A.COMPETENT_TAX_AUTHORITY AS ISSUED_BY,\n" +
                            "A.BALANCE AS AMOUNT,\n" +
                            "A.BALANCE_NEW AS NEW_AMOUNT,\n" +
                            "A.JOBID, \n" +
                            "'1' AS ISADD,a.creditcode\n" +
                            "FROM \n" +
                            "(\n" +
                            "    select del.creditcode,t.* \n" +
                            "    from a_dc_ep_ods.zs_t_tax_arrears t  \n" +
                            "    inner join a_dc_ep_ods.zs_t_tax_arrears_del del on del.entid = t.entid\n" +
                            "    where del.creditcode is not null and del.creditcode != '' \n" +
                            ") A LEFT JOIN \n" +
                            "(\n" +
                            "    select c.credit_code,o.*\n" +
                            "    from a_dc_ep_dwi.t_tax_owe o\n" +
                            "    inner join a_dc_ep_ods.t_eci_company c on c.company_id = o.company_id\n" +
                            "    where c.credit_code is not null and c.credit_code != '' \n" +
                            ") B \n" +
                            "ON UPPER(TRIM(B.CREDIT_CODE)) = UPPER(TRIM(A.CREDITCODE)) \n" +
                            "AND DATE_FORMAT(IF(A.NOTICE_DATE = '' OR A.NOTICE_DATE IS NULL,'1900-01-01',A.NOTICE_DATE),'yyyy-MM-dd')\n" +
                            "=\n" +
                            "DATE_FORMAT(IF(B.PUBLISH_DATE = '' OR B.PUBLISH_DATE IS NULL,'1900-01-01',B.PUBLISH_DATE),'yyyy-MM-dd')\n" +
                            "AND UPPER(TRIM(A.TAX_CATEGORY)) = UPPER(TRIM(B.TITLE))\n" +
                            "WHERE B.CREDIT_CODE IS NULL");
            insert_data.createOrReplaceTempView("tmp_t_tax_owe1");
            spark.sql("select a.id,a.KEY_NO,if(b.company_id is not null,b.company_id,a.company_id) company_id,a.COMPANY_NAME,a.TAX_OWE_ID,a.title,a.PUBLISH_DATE,a.ISSUED_BY,a.AMOUNT,a.NEW_AMOUNT,a.JOBID,a.ISADD " +
                    "from (select * from tmp_t_tax_owe1 where length(trim(creditcode))=18) as a left join a_dc_ep_ods.t_eci_company as b on " +
                    "a.creditcode = b.credit_code " +
                    "union all " +
                    "select a.id,a.KEY_NO,if(b.company_id is not null,b.company_id,a.company_id) company_id,a.COMPANY_NAME,a.TAX_OWE_ID,a.title,a.PUBLISH_DATE,a.ISSUED_BY,a.AMOUNT,a.NEW_AMOUNT,a.JOBID,a.ISADD " +
                    "from (select * from tmp_t_tax_owe1 where length(trim(creditcode))!=18) as a left join a_dc_ep_ods.t_eci_company as b on " +
                    "regexp_replace(trim(a.COMPANY_NAME),'([(（）)])','') = regexp_replace(trim(b.COMPANY_NAME),'([(（）)])','')").createOrReplaceTempView("tmp_t_tax_owe11");
            spark.sql("insert into table a_dc_ep_incr.t_tax_owe select * from tmp_t_tax_owe11");

            // 3.DELETE
            // SQL逻辑：定义 delete_data 集合
            //          1.从 xy_ods 取得企查查全集 为集合A
            //          2.从 dc_ep_ods 取得( 以_del结尾表为主表，left join原始表，找到删除的entid)  中数删除 全集 为集合B
            //          3.集合A 和 集合B 通过统一社会信用代码，做INNER JOIN，注意要 UPPER 和 TRIM 函数，再筛选企查查侧为空的 为集合C
            //                  取得企查查侧全部字段 并加
            //                  JOB_ID, "-1" as isadd 两个字段是要增加的
            //          4.delete_data.createOrReplaceTempView 完成插入
            //          9.同步到目标a_dc_ep_incr库表
            Dataset<Row> delete_data = spark.sql("" +
                    "SELECT \n" +
                    "A.ID,\n" +
                    "A.KEY_NO,\n" +
                    "A.COMPANY_ID,\n" +
                    "A.COMPANY_NAME,\n" +
                    "A.TAX_OWE_ID,\n" +
                    "A.TITLE,\n" +
                    "A.PUBLISH_DATE,\n" +
                    "A.ISSUED_BY,\n" +
                    "A.AMOUNT,\n" +
                    "A.NEW_AMOUNT,\n" +
                    "B.JOBID,\n" +
                    "'-1' AS ISADD\n" +
                    "FROM\n" +
                    "(\n" +
                    "    select c.credit_code,o.*\n" +
                    "    from a_dc_ep_dwi.t_tax_owe o\n" +
                    "    inner join a_dc_ep_ods.t_eci_company c on c.company_id = o.company_id\n" +
                    "    where c.credit_code is not null and c.credit_code != '' \n" +
                    ") A\n" +
                    "INNER JOIN\n" +
                    "(\n" +
                    "    select a.creditcode,a.jobid from (\n" +
                    "        select del.entid,del.creditcode,del.jobid \n" +
                    "        from a_dc_ep_ods.zs_t_tax_arrears_del del\n" +
                    "        left join a_dc_ep_ods.zs_t_tax_arrears ta on ta.entid = del.entid\n" +
                    "        where ta.entid is null\n" +
                    "    ) a  \n" +
                    "    where a.creditcode is not null and a.creditcode != '' \n" +
                    ") B\n" +
                    "ON UPPER(TRIM(B.creditcode)) = UPPER(TRIM(A.credit_code))");
            delete_data.createOrReplaceTempView("tmp_t_tax_owe2");
            spark.sql("insert into table a_dc_ep_incr.t_tax_owe select * from tmp_t_tax_owe2");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_tax_owe");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
        spark.stop();
    }
}
